package cn.yomigaeri.signal.consumer;

import cn.yomigaeri.common.entity.dto.SignalDTO;
import cn.yomigaeri.common.entity.po.Signal;
import cn.yomigaeri.signal.mapper.SignalMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

/**
 * @ClassName: SignalConsumer
 * @Package: cn.yomigaeri.signal.consumer
 * @Description:
 * @Author Yomigaeri
 * @Create 2025/6/4 21:45
 * @Version 1.0
 */
@Service
@Slf4j
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = "signalToSave1",
        consumerGroup = "signal-consumer"
)
public class SignalConsumer1 implements RocketMQListener<SignalDTO> {

    private final SignalMapper signalMapper;

    // 缓存消息列表
    private final List<Signal> signalList = new ArrayList<>();


    @Override
    //幂等性处理：如果消息可能重复，建议在插入前检查是否已存在相同的消息。
    //事务管理：如果你希望每次插入要么全部成功要么全部失败，可以在业务逻辑中加入事务控制。
    //性能优化：根据实际需求调整批处理大小（如 100 条）以平衡内存占用和吞吐量
    public void onMessage(SignalDTO signalDTO) {
        synchronized (signalList) {
            try {
                Signal signal = new Signal();
                BeanUtils.copyProperties(signalDTO, signal);
                // 添加消息到缓存
                signalList.add(signal);

                // 达到100条时批量插入
                if (signalList.size() >= 3) {
                    signalMapper.batchInsertSignals1(signalList);
                    signalList.clear(); // 清空缓存
                }
            } catch (Exception e) {
                log.error("Failed to process message: {}", e.getMessage(), e);
                // 可选：将失败的消息重新放回队列或进行补偿处理
            }
        }
    }
}
